package org.example.api.process;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.example.api.bean.SensorReading;

/**
 * @author huangqihan
 * @date 2021/2/25
 */
public class KeyedProcessFuncApp {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        String host = "localhost";
        int port = 7777;

        // Use NetCat to read the data source. nc -lp 7777
        DataStreamSource<String> inputStream = env.socketTextStream(host, port);

        SingleOutputStreamOperator<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        SingleOutputStreamOperator<String> result = dataStream
                .keyBy("id")
                .process(new ProcessFunc(10));

        result.print();
        env.execute();
    }

    public static class ProcessFunc extends KeyedProcessFunction<Tuple, SensorReading, String> {

        private Integer interval;

        public ProcessFunc(Integer interval) {
            this.interval = interval;
        }

        ValueState<Double> lastTempState;
        ValueState<Long> timeTsState;

        @Override
        public void open(Configuration parameters) throws Exception {
            timeTsState = getRuntimeContext().getState(new ValueStateDescriptor<>("tsTimer", Long.class));
            lastTempState = getRuntimeContext().getState(new ValueStateDescriptor<>("lastTemp", Double.class));
        }

        /**
         * 监控10秒内温度，如果持续上升，触发报警
         */
        @Override
        public void processElement(SensorReading sensorReading, Context context, Collector<String> collector) throws Exception {
            Double lastTempStateValue = lastTempState.value();
            Long timeTsStateValue = timeTsState.value();

            if (sensorReading.getTemperature() > lastTempStateValue && timeTsStateValue == null) {
                // fixme some bugs
                long ts = context.timerService().currentProcessingTime() + interval * 1000L;
                timeTsState.update(ts);
                context.timerService().registerEventTimeTimer(ts);
            } else if (sensorReading.getTemperature()<lastTempStateValue){
                context.timerService().deleteProcessingTimeTimer(timeTsStateValue);
                timeTsState.clear();
            }

            lastTempState.update(sensorReading.getTemperature());
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
            out.collect("传感器"+ctx.getCurrentKey().getField(0)+"温度连续"+interval+"s上升.");
        }
    }
}
